package com.jch.ts;

import io.netty.channel.*;
import io.netty.util.internal.shaded.org.jctools.queues.MessagePassingQueue;

import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;

public abstract class TSCommonHandler extends SimpleChannelInboundHandler<String> {


    /**
     * 发送标识
     */
    protected boolean sendFlag;
    /**
     * 当前任务
     */
    private Runnable currentTask;
    /**
     * 消费者消息长度
     */
    private AtomicLong consumerMsgLength = new AtomicLong();
    /**
     * 当前进度
     */
    private long priorProgress;


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        sendData(ctx);
        // 监控线程启动
        new Thread(currentTask).start();
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        // 实现监控线程
        currentTask = () -> {
            while (true) {
                try {
                    Thread.sleep(500);
                    long length = consumerMsgLength.getAndSet(0);
                    if (0 == length) continue;
                    System.out.println("当前发送速率（bytes/s): " + length);
                } catch (InterruptedException ignored) {
                }
            }
        };
        super.handlerAdded(ctx);
    }

    protected ChannelProgressivePromise getChannelProgressivePromise(ChannelHandlerContext ctx, MessagePassingQueue.Consumer<ChannelProgressiveFuture> completedAction) {
        ChannelProgressivePromise channelProgressivePromise = ctx.newProgressivePromise();
        channelProgressivePromise.addListener(new ChannelProgressiveFutureListener() {
            @Override
            public void operationProgressed(ChannelProgressiveFuture future, long progress, long total) throws Exception {
                consumerMsgLength.addAndGet(progress - priorProgress);
                System.out.println("consumerMsgLength：" + consumerMsgLength.get());
                priorProgress = progress;
                System.out.println("当前进度：" + priorProgress + " 总量：" + total);
            }

            @Override
            public void operationComplete(ChannelProgressiveFuture future) throws Exception {
                sendFlag = false;
                if (future.isSuccess()) {
                    System.out.println("消息发送成功。");
                    priorProgress -= 10;
                    Optional.ofNullable(completedAction).ifPresent(action -> action.accept(future));
                } else {
                    System.out.println("消息发送失败.");
                    future.cause().printStackTrace();
                }
            }
        });
        return channelProgressivePromise;
    }

    // 发送数据
    protected abstract void sendData(ChannelHandlerContext ctx);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println("收到消息：" + msg);
    }
}
